package com.gitee.melin.bee.core.jdbc.dialect;

import static com.gitee.melin.bee.core.jdbc.Constants.COLUMN_COMMENT;
import static com.gitee.melin.bee.core.jdbc.Constants.COLUMN_NAME;
import static com.gitee.melin.bee.core.jdbc.Constants.COLUMN_TYPE;
import static com.gitee.melin.bee.core.jdbc.Constants.CREATE_TIME;
import static com.gitee.melin.bee.core.jdbc.Constants.DATA_LENGTH;
import static com.gitee.melin.bee.core.jdbc.Constants.DATA_SIZE;
import static com.gitee.melin.bee.core.jdbc.Constants.DATA_TYPE;
import static com.gitee.melin.bee.core.jdbc.Constants.DATETIME_PRECISION;
import static com.gitee.melin.bee.core.jdbc.Constants.DEFAULT_VALUE;
import static com.gitee.melin.bee.core.jdbc.Constants.IS_NULLABLE;
import static com.gitee.melin.bee.core.jdbc.Constants.NUMERIC_PRECISION;
import static com.gitee.melin.bee.core.jdbc.Constants.NUMERIC_SCALE;
import static com.gitee.melin.bee.core.jdbc.Constants.SQL2003_KEYWORDS;
import static com.gitee.melin.bee.core.jdbc.Constants.TABLE_COMMENT;
import static com.gitee.melin.bee.core.jdbc.Constants.TABLE_ENGINE;
import static com.gitee.melin.bee.core.jdbc.Constants.TABLE_NAME;
import static com.gitee.melin.bee.core.jdbc.Constants.TABLE_PARTITION;
import static com.gitee.melin.bee.core.jdbc.Constants.TABLE_ROWS;
import static com.gitee.melin.bee.core.jdbc.Constants.TABLE_TYPE;
import static com.gitee.melin.bee.core.jdbc.Constants.UPDATE_TIME;
import static com.gitee.melin.bee.core.jdbc.enums.DataSourceType.*;

import com.gitee.melin.bee.core.jdbc.ConnectionFactory;
import com.gitee.melin.bee.core.jdbc.ConnectionProxy;
import com.gitee.melin.bee.core.jdbc.DriverConnectionFactory;
import com.gitee.melin.bee.core.jdbc.JdbcInvocationHandler;
import com.gitee.melin.bee.core.jdbc.NamedParameterStatement;
import com.gitee.melin.bee.core.jdbc.QueryResult;
import com.gitee.melin.bee.core.jdbc.enums.CompatibleMode;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.*;
import com.gitee.melin.bee.util.JdbcUtils;
import com.gitee.melin.bee.util.ThreadUtils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import java.lang.reflect.Proxy;
import java.sql.*;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class AbstractJdbcDialect implements JdbcDialect {

    private static final Logger LOG = LoggerFactory.getLogger(AbstractJdbcDialect.class);

    protected ConnectionInfo connectionInfo;

    protected ConnectionFactory connectionFactory;

    private volatile PreparedStatement preparedStatement;

    private Connection connection;

    protected CompatibleMode compatibleMode;

    public AbstractJdbcDialect(ConnectionInfo connectionInfo) {
        if (connectionInfo == null) {
            return;
        }

        this.connectionInfo = connectionInfo;

        Map<String, String> properties = connectionInfo.getProperties();
        if (properties == null) {
            properties = Maps.newHashMap();
        }

        connectionProperties(properties);
        connectionFactory = createConnectionFactory(properties);

        if (connectionInfo.getCompatibleMode() != null) {
            compatibleMode = connectionInfo.getCompatibleMode();
        } else {
            compatibleMode = queryCompatibleMode();
        }
    }

    public AbstractJdbcDialect(Connection connection) {
        this.connection = connection;
    }

    private CompatibleMode queryCompatibleMode() {
        // OceanBase需判断租户信息
        CompatibleMode compatibleMode = null;
        if (DataSourceType.OCEANBASE == connectionInfo.getDataSourceType()) {
            QueryResult queryResult = query("show variables like 'ob_compatibility_mode';");
            LinkedList<LinkedList<Object>> rows = queryResult.getRows();
            String mode = rows.get(0).get(1).toString();
            if ("oracle".equalsIgnoreCase(mode)) {
                LOG.info("OceanBase compatible mode is oracle");
                compatibleMode = CompatibleMode.ORACLE;
            } else {
                compatibleMode = CompatibleMode.MYSQL;
            }
        }

        return compatibleMode;
    }

    protected void connectionProperties(Map<String, String> connectionProperties) {}

    public ConnectionFactory createConnectionFactory(Map<String, String> properties) {
        try {
            return new DriverConnectionFactory(
                    getDriver(), getJdbcUrl(), connectionInfo, properties, this.getNativeDataSource());
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    protected String getJdbcUrl() {
        return connectionInfo.getUrl();
    }

    protected abstract Driver getDriver() throws SQLException;

    public Connection getConnection() throws SQLException {
        if (connectionFactory != null) {
            return connectionFactory.openConnection();
        } else {
            return createConnectionProxy(connection); // 避免关闭连接
        }
    }

    private Connection createConnectionProxy(Connection con) {
        return (Connection) Proxy.newProxyInstance(
                ConnectionProxy.class.getClassLoader(),
                new Class<?>[] {ConnectionProxy.class},
                new JdbcInvocationHandler(con));
    }

    @Override
    public MetaDataSource testConnection(ConnectionInfo connectionInfo) {
        try (Connection connection = DriverManager.getConnection(
                        this.getJdbcUrl(), connectionInfo.getUsername(), connectionInfo.getPassword());
                PreparedStatement statement = connection.prepareStatement(testConnectionSql());
                ResultSet resultSet = statement.executeQuery()) {

            DatabaseVersion version = getDatabaseVersion(connection);
            MetaDataSource dataSourceInfo = new MetaDataSource(connection.getMetaData(), version);
            if (compatibleMode != null) {
                dataSourceInfo.setCompatibleMode(compatibleMode);
            }
            return dataSourceInfo;
        } catch (SQLException e) {
            throw new RuntimeException("Test connection failed: " + e.getMessage(), e);
        }
    }

    protected String testConnectionSql() {
        return "select 1";
    }

    @Override
    public List<MetaProcedure> getProcedures(String catalogName, String schemaPattern, String functionNamePattern) {
        try (Connection connection = this.getConnection();
                ResultSet resultSet =
                        connection.getMetaData().getProcedures(catalogName, schemaPattern, functionNamePattern)) {

            List<MetaProcedure> list = Lists.newArrayList();
            while (resultSet.next()) {
                String funcDatabase = resultSet.getString("PROCEDURE_CAT");
                String funcSchema = resultSet.getString("PROCEDURE_SCHEM");
                String funcName = resultSet.getString("PROCEDURE_NAME");
                String funcType = resultSet.getString("PROCEDURE_TYPE");
                String comment = resultSet.getString("REMARKS");
                MetaProcedure metaProcedure = new MetaProcedure();
                metaProcedure.setProcDatabase(funcDatabase);
                metaProcedure.setProcSchema(funcSchema);
                metaProcedure.setProcName(funcName);
                metaProcedure.setProcType(funcType);
                metaProcedure.setComment(comment);
                list.add(metaProcedure);
            }
            return list;
        } catch (SQLException e) {
            throw new RuntimeException("get sql functions failed: " + e.getMessage(), e);
        }
    }

    @Override
    public List<MetaFunction> getFunctions(String catalogName, String schemaPattern, String functionNamePattern) {
        try (Connection connection = this.getConnection();
                ResultSet resultSet =
                        connection.getMetaData().getFunctions(catalogName, schemaPattern, functionNamePattern)) {

            List<MetaFunction> list = Lists.newArrayList();
            while (resultSet.next()) {
                String funcDatabase = resultSet.getString("FUNCTION_CAT");
                String funcSchema = resultSet.getString("FUNCTION_SCHEM");
                String funcName = resultSet.getString("FUNCTION_NAME");
                String funcType = resultSet.getString("FUNCTION_TYPE");
                String comment = resultSet.getString("REMARKS");
                MetaFunction metaFunc = new MetaFunction();
                metaFunc.setFuncDatabase(funcDatabase);
                metaFunc.setFuncSchema(funcSchema);
                metaFunc.setFuncName(funcName);
                metaFunc.setFuncType(funcType);
                metaFunc.setComment(comment);
                list.add(metaFunc);
            }
            return list;
        } catch (SQLException e) {
            throw new RuntimeException("get sql functions failed: " + e.getMessage(), e);
        }
    }

    @Override
    public List<MetaTypeInfo> getMetaTypeInfos() {
        try (Connection connection = this.getConnection();
                ResultSet resultSet = connection.getMetaData().getTypeInfo()) {

            List<MetaTypeInfo> list = Lists.newArrayList();
            while (resultSet.next()) {
                String typeName = resultSet.getString("TYPE_NAME");
                Integer dataType = resultSet.getInt("DATA_TYPE");
                Integer precision = resultSet.getInt("PRECISION");
                Short minScale = resultSet.getShort("MINIMUM_SCALE");
                Short maxScale = resultSet.getShort("MAXIMUM_SCALE");
                MetaTypeInfo typeInfo = new MetaTypeInfo();
                typeInfo.setTypeName(typeName);
                typeInfo.setDataType(dataType);
                typeInfo.setPrecision(precision);
                typeInfo.setMinScale(minScale);
                typeInfo.setMaxScale(maxScale);
                list.add(typeInfo);
            }
            return list;
        } catch (SQLException e) {
            throw new RuntimeException("get type names failed: " + e.getMessage(), e);
        }
    }

    @Override
    public List<String> getSystemFunctions() {
        try (Connection connection = this.getConnection()) {
            List<String> allFunctions = Lists.newArrayList();
            String functions = connection.getMetaData().getSystemFunctions();
            if (StringUtils.isNotBlank(functions)) {
                allFunctions.add(functions);
            }
            functions = connection.getMetaData().getNumericFunctions();
            if (StringUtils.isNotBlank(functions)) {
                allFunctions.add(functions);
            }
            functions = connection.getMetaData().getStringFunctions();
            if (StringUtils.isNotBlank(functions)) {
                allFunctions.add(functions);
            }
            functions = connection.getMetaData().getTimeDateFunctions();
            if (StringUtils.isNotBlank(functions)) {
                allFunctions.add(functions);
            }

            String funcs = StringUtils.join(allFunctions, ",");
            return Lists.newArrayList(StringUtils.split(funcs, ","));
        } catch (SQLException e) {
            throw new RuntimeException("get sql keywords failed: " + e.getMessage(), e);
        }
    }

    @Override
    public List<String> getSqlKeywords() {
        try (Connection connection = this.getConnection()) {
            String[] keywords = StringUtils.split(connection.getMetaData().getSQLKeywords(), ",");
            Set<String> set = Arrays.stream(keywords).map(String::toUpperCase).collect(Collectors.toSet());
            set.addAll(SQL2003_KEYWORDS);
            List<String> result = Lists.newArrayList(set);
            Collections.sort(result);
            return result;
        } catch (SQLException e) {
            throw new RuntimeException("get sql keywords failed: " + e.getMessage(), e);
        }
    }

    @Override
    public DatabaseVersion getDatabaseVersion() {
        try (Connection connection = this.getConnection()) {
            return getDatabaseVersion(connection);
        } catch (SQLException e) {
            throw new RuntimeException("Test connection failed: " + e.getMessage(), e);
        }
    }

    private DatabaseVersion getDatabaseVersion(Connection connection) {
        try {
            DatabaseMetaData metaData = connection.getMetaData();

            try {
                return new DatabaseVersion(metaData.getDatabaseProductVersion());
            } catch (Throwable e) {
                try {
                    int major = interpretVersion(metaData.getDatabaseMajorVersion());
                    int minor = interpretVersion(metaData.getDatabaseMinorVersion());
                    return new DatabaseVersion(major, minor, 0);
                } catch (IllegalArgumentException e1) {
                    LOG.debug("Can't determine database version. Use default");
                    return new DatabaseVersion(0, 0, 0);
                }
            }
        } catch (SQLException e) {
            throw new RuntimeException("Test connection failed: " + e.getMessage(), e);
        }
    }

    @Override
    public List<String> getDatabases() {
        try (Connection connection = this.getConnection();
                ResultSet resultSet = connection.getMetaData().getCatalogs()) {

            List<String> catalogs = Lists.newArrayList();
            while (resultSet.next()) {
                String catalogName = resultSet.getString(1);
                catalogs.add(catalogName);
            }
            return catalogs;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    public List<String> getSchemas() {
        try (Connection connection = this.getConnection()) {
            return listSchemas(connection);
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    protected List<String> listSchemas(Connection connection) {
        try (ResultSet resultSet = connection.getMetaData().getSchemas(connection.getCatalog(), null)) {
            List<String> schemaNames = Lists.newArrayList();
            while (resultSet.next()) {
                String schemaName = resultSet.getString("TABLE_SCHEM");
                // skip internal schemas
                if (filterSchema(schemaName)) {
                    schemaNames.add(schemaName);
                }
            }
            return schemaNames;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    protected boolean filterSchema(String schemaName) {
        return !schemaName.equalsIgnoreCase("information_schema");
    }

    @Override
    public List<String> getTableNames(String schemaName) {
        List<String> tables = new ArrayList<>();
        String catalog = schemaName;
        if (this.getDataSourceType() == SQLSERVER || this.getDataSourceType() == DAMENG) {
            catalog = null;
        }

        try (Connection connection = this.getConnection()) {
            ResultSet resultSet = connection
                    .getMetaData()
                    .getTables(catalog, schemaName, "%", new String[] {"TABLE", "VIEW", "MATERIALIZED VIEW"});

            while (resultSet.next()) {
                tables.add(resultSet.getString("TABLE_NAME"));
            }
            return tables;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    public List<SimpleMetaTable> getSimpleTables(String schemaName) {
        List<SimpleMetaTable> tables = new ArrayList<>();
        try (Connection connection = this.getConnection()) {
            ResultSet resultSet = connection
                    .getMetaData()
                    .getTables(schemaName, null, "%", new String[] {"TABLE", "VIEW", "MATERIALIZED VIEW"});

            while (resultSet.next()) {
                SimpleMetaTable table = new SimpleMetaTable();
                table.setTableName(resultSet.getString("TABLE_NAME"));
                table.setComment(resultSet.getString("REMARKS"));

                String type = resultSet.getString("TABLE_TYPE");
                if (StringUtils.equalsIgnoreCase(type, "VIEW")) {
                    table.setTableType(BeeTableType.VIEW);
                } else if (StringUtils.equalsIgnoreCase(type, "TABLE")) {
                    table.setTableType(BeeTableType.TABLE);
                } else if (StringUtils.equalsIgnoreCase(type, "Materialized View")) {
                    table.setTableType(BeeTableType.MATERIALIZED_VIEW);
                } else {
                    throw new IllegalArgumentException("simple table type is invalid: " + type);
                }
                tables.add(table);
            }
            return tables;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    public List<MetaTable> getSchemaTables(String schemaName) {
        return this.getSchemaTables(schemaName, null);
    }

    protected List<MetaTable> getSchemaTables(String schemaName, String tableName) {
        try (Connection connection = this.getConnection()) {
            String sql = this.getSelectTablesSql(tableName);
            NamedParameterStatement preparedStatement = null;
            ResultSet resultSet = null;
            try {
                preparedStatement = new NamedParameterStatement(connection, sql);
                if (connectionInfo != null && connectionInfo.getQueryTimeout() != null) {
                    preparedStatement.setQueryTimeout(connectionInfo.getQueryTimeout());
                }

                preparedStatement.setString("schemaName", schemaName);
                if (StringUtils.isNotBlank(tableName)) {
                    preparedStatement.setString("tableName", tableName);
                }

                resultSet = preparedStatement.executeQuery();

                List<MetaTable> tables = new ArrayList<>();
                while (resultSet.next()) {
                    MetaTable metaTable = buildTable(resultSet);
                    if (metaTable != null) {
                        metaTable.setSchemaName(schemaName);
                        tables.add(metaTable);
                    }
                }
                return tables;
            } finally {
                if (preparedStatement != null) {
                    try {
                        preparedStatement.close();
                    } catch (SQLException ex) {
                        LOG.trace("Could not close JDBC ResultSet", ex);
                    } catch (Throwable ex) {
                        LOG.trace("Unexpected exception on closing JDBC ResultSet", ex);
                    }
                }
                JdbcUtils.closeResultSet(resultSet);
            }
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    protected String getSelectTablesSql(String tableName) {
        throw new UnsupportedOperationException();
    }

    protected MetaTable buildTable(ResultSet resultSet) throws SQLException {
        String tableName = resultSet.getString(TABLE_NAME);
        try {
            if (filterTable(tableName)) {
                MetaTable tableInfo = new MetaTable();

                tableInfo.setTableName(tableName);
                tableInfo.setComment(resultSet.getString(TABLE_COMMENT));
                tableInfo.setEngine(resultSet.getString(TABLE_ENGINE));

                String tableType = resultSet.getString(TABLE_TYPE);
                tableType = StringUtils.trim(tableType);
                BeeTableType beeTableType = getTableType(tableType);
                tableInfo.setTableType(beeTableType);

                String partition = resultSet.getString(TABLE_PARTITION);
                boolean isPart = isParttitionTable(partition);
                tableInfo.setPartition(isPart);

                tableInfo.setNumRows(resultSet.getLong(TABLE_ROWS));
                tableInfo.setDataSize(resultSet.getLong(DATA_SIZE));
                tableInfo.setCreateTime(resultSet.getTimestamp(CREATE_TIME));
                tableInfo.setUpdateTime(resultSet.getTimestamp(UPDATE_TIME));
                return tableInfo;
            }
        } catch (Exception e) {
            throw new RuntimeException(tableName + " build column error: " + e.getMessage());
        }

        return null;
    }

    @Override
    public LinkedHashSet<String> getPrimaryKeys(String schemaName, String tableName) {
        try (Connection connection = this.getConnection()) {
            return getPrimaryKeys(connection, schemaName, tableName);
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }

    private LinkedHashSet<String> getPrimaryKeys(Connection connection, String schemaName, String tableName)
            throws SQLException {
        DatabaseMetaData metaData = connection.getMetaData();
        String catalog = connection.getCatalog();
        if (StringUtils.isBlank(catalog)) {
            catalog = null;
        }

        ResultSet rs = metaData.getPrimaryKeys(catalog, schemaName, tableName);

        // seq -> column name
        LinkedHashSet<String> primaryKeyColumns = new LinkedHashSet<>();
        String pkName = null;
        while (rs.next()) {
            String columnName = rs.getString("COLUMN_NAME");
            // all the PK_NAME should be the same
            pkName = rs.getString("PK_NAME");
            int keySeq = rs.getInt("KEY_SEQ");
            // KEY_SEQ is 1-based index
            primaryKeyColumns.add(columnName);
        }
        return primaryKeyColumns;
    }

    @Override
    public MetaTable getSchemaTable(String schemaName, String tableName, boolean containColumn) {
        List<MetaTable> tables = this.getSchemaTables(schemaName, tableName);
        if (tables.isEmpty()) {
            return null;
        } else if (tables.size() > 1) {
            String msg = String.format("table %s.%s has more than one", schemaName, tableName);
            throw new RuntimeException(msg);
        } else {
            MetaTable tableInfo = tables.get(0);
            if (containColumn) {
                List<MetaColumn> columns = this.getSchemaColumns(schemaName, tableName);
                tableInfo.setColumns(columns);
            }
            return tableInfo;
        }
    }

    protected boolean filterTable(String tableName) {
        return true;
    }

    @Override
    public List<MetaColumn> getSchemaColumns(String schemaName, String tableName) {
        try (Connection connection = this.getConnection()) {
            Map<String, MetaColumn> columnMap = Maps.newLinkedHashMap();
            try {
                DatabaseMetaData metaData = connection.getMetaData();
                ResultSet results = metaData.getColumns(null, schemaName, tableName, null);
                LinkedHashSet<String> primaryKeys = getPrimaryKeys(connection, schemaName, tableName);
                while (results.next()) {
                    String columnName = results.getString("COLUMN_NAME");
                    int jdbcTypeInt = results.getInt("DATA_TYPE");
                    JDBCType jdbcType = JDBCType.valueOf(jdbcTypeInt);
                    String dataType = results.getString("TYPE_NAME");

                    MetaColumn metaColumn = new MetaColumn();
                    metaColumn.setColumnName(columnName);
                    metaColumn.setDataType(dataType);
                    metaColumn.setJdbcType(jdbcType);

                    if (primaryKeys.contains(columnName)) {
                        metaColumn.setPrimaryKey(true);
                    }

                    if (JDBCType.DOUBLE == jdbcType
                            || JDBCType.FLOAT == jdbcType
                            || JDBCType.DECIMAL == jdbcType
                            || JDBCType.NUMERIC == jdbcType) {
                        metaColumn.setPrecision(results.getLong("COLUMN_SIZE"));
                        metaColumn.setScale(results.getInt("DECIMAL_DIGITS"));
                    } else {
                        metaColumn.setLength(results.getLong("COLUMN_SIZE"));
                        metaColumn.setMaxLength(results.getLong("CHAR_OCTET_LENGTH"));
                    }

                    metaColumn.setNullable(results.getBoolean("NULLABLE"));
                    metaColumn.setComment(results.getString("REMARKS"));
                    metaColumn.setAutoIncrement(results.getBoolean("IS_AUTOINCREMENT"));
                    metaColumn.setPosition(results.getInt("ORDINAL_POSITION"));
                    metaColumn.setDefaultValue(results.getObject("COLUMN_DEF"));
                    columnMap.put(columnName, metaColumn);
                }
            } catch (SQLException e) {
                throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
            }

            String sql = this.getSelectColumnsSql();
            if (sql != null) {
                NamedParameterStatement preparedStatement = new NamedParameterStatement(connection, sql);
                preparedStatement.setString("schemaName", schemaName);
                preparedStatement.setString("tableName", tableName);

                List<MetaColumn> columns = new ArrayList<>();
                try (ResultSet resultSet = preparedStatement.executeQuery()) {
                    while (resultSet.next()) {
                        String columnName = resultSet.getString(COLUMN_NAME);
                        MetaColumn metaColumn = columnMap.get(columnName);
                        metaColumn = buildColumn(metaColumn, resultSet);
                        columns.add(metaColumn);
                    }
                } finally {
                    try {
                        preparedStatement.close();
                    } catch (SQLException ex) {
                        LOG.trace("Could not close JDBC ResultSet", ex);
                    } catch (Throwable ex) {
                        LOG.trace("Unexpected exception on closing JDBC ResultSet", ex);
                    }
                }
                return columns;
            } else {
                return Lists.newArrayList(columnMap.values());
            }
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    protected String getSelectColumnsSql() {
        return null;
    }

    protected MetaColumn buildColumn(MetaColumn column, ResultSet resultSet) throws SQLException {
        // e.g. tinyint(1) unsigned
        String columnType = resultSet.getString(COLUMN_TYPE);
        // e.g. tinyint
        String dataType = resultSet.getString(DATA_TYPE).toUpperCase();
        String comment = resultSet.getString(COLUMN_COMMENT);
        Object defaultValue = resultSet.getObject(DEFAULT_VALUE);
        String isNullableStr = resultSet.getString(IS_NULLABLE);
        boolean isNullable = isNullableStr.equals("YES");
        // e.g. `decimal(10, 2)` is 10
        long numberPrecision = resultSet.getInt(NUMERIC_PRECISION);
        // e.g. `decimal(10, 2)` is 2
        int numberScale = resultSet.getInt(NUMERIC_SCALE);
        // e.g. `varchar(10)` is 40
        long dataLength = resultSet.getLong(DATA_LENGTH);
        // e.g. `timestamp(3)` is 3
        int timePrecision = resultSet.getInt(DATETIME_PRECISION);
        boolean unsigned = columnType.toLowerCase(Locale.ROOT).contains("unsigned");

        if (StringUtils.isBlank(column.getColumnType())) {
            column.setColumnType(columnType);
        }
        if (StringUtils.isBlank(column.getDataType())) {
            column.setDataType(dataType);
        }

        long length = Math.max(dataLength, numberPrecision);
        if (column.getLength() == null || length > 0) {
            column.setLength(length);
        }
        column.setPrecision(numberPrecision);
        int scala = Math.max(numberScale, timePrecision);
        if (column.getScale() == null || scala > 0) {
            column.setScale(Math.max(numberScale, timePrecision));
        }
        column.setNullable(isNullable);
        if (column.getDefaultValue() == null) {
            column.setDefaultValue(defaultValue);
        }

        if (StringUtils.isBlank(column.getComment())) {
            column.setComment(comment);
        }
        column.setUnsigned(unsigned);
        return column;
    }

    @Override
    public String getCreateTableScript(BeeTableType type, String schemaName, String tableName) {
        String sql = createTableSql(type, schemaName, tableName);

        try (Connection connection = this.getConnection();
                PreparedStatement dbStat = connection.prepareStatement(sql)) {
            try (ResultSet dbResult = dbStat.executeQuery()) {
                if (dbResult.next()) {
                    String colName = getDDLColumnName(type);
                    return dbResult.getString(colName);
                } else {
                    return "DDL is not available";
                }
            }
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    protected String createTableSql(BeeTableType type, String schemaName, String tableName) {
        if (BeeTableType.TABLE == type) {
            return "show create table " + schemaName + "." + tableName;
        } else if (BeeTableType.VIEW == type) {
            return "show create view " + schemaName + "." + tableName;
        } else {
            return "show create materialized view " + schemaName + "." + tableName;
        }
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String sql, int maxRecords) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> query(sql, maxRecords), executor);
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String schema, String sql, int maxRecords) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> query(schema, sql, maxRecords), executor);
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String schema, String sql) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> query(schema, sql, -1), executor);
    }

    @Override
    public CompletableFuture<QueryResult> asyncQuery(String sql) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> query(sql, -1), executor);
    }

    @Override
    public CompletableFuture<Boolean> asyncExecute(String schema, String sql) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> execute(schema, sql), executor);
    }

    @Override
    public CompletableFuture<Boolean> asyncExecute(String sql) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> execute(sql), executor);
    }

    @Override
    public CompletableFuture<Integer> asyncExecuteUpdate(SchemaIdentifier schema, String sql, Object... params) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> executeUpdate(schema, sql, params), executor);
    }

    @Override
    public CompletableFuture<Integer> asyncExecuteUpdate(String sql, Object... params) {
        ExecutorService executor = ThreadUtils.newDaemonSingleThreadExecutor("executeSql");
        return CompletableFuture.supplyAsync(() -> executeUpdate(sql, params), executor);
    }

    @Override
    public QueryResult query(String sql) {
        return query(null, sql, -1);
    }

    @Override
    public QueryResult query(String schema, String sql) {
        return query(schema, sql, -1);
    }

    @Override
    public QueryResult query(String sql, int maxRecords) {
        return this.query(null, sql, maxRecords);
    }

    // https://simplefeatures.wordpress.com/2011/09/30/jdbc-tweaks-to-enable-streaming-result-sets/
    @Override
    public QueryResult query(String schema, String sql, int maxRecords) {
        String jdbcUrl = this.getJdbcUrl();
        if (SQLSERVER == this.getDataSourceType()) {
            jdbcUrl += ";selectMethod=cursor";
        }

        Connection conn = null;
        ResultSet result = null;
        try {
            conn = DriverManager.getConnection(jdbcUrl, connectionInfo.getUsername(), connectionInfo.getPassword());

            if (schema != null) {
                if (MYSQL == connectionInfo.getDataSourceType() || STARROCKS == connectionInfo.getDataSourceType()) {
                    conn.setCatalog(schema);
                } else {
                    conn.setSchema(schema);
                }
            }

            try {
                conn.setAutoCommit(false);
            } catch (SQLFeatureNotSupportedException ignore) {
            }

            preparedStatement = conn.prepareStatement(sql);
            if (MYSQL == this.getDataSourceType()) {
                preparedStatement.setFetchSize(Integer.MIN_VALUE);
            } else {
                preparedStatement.setFetchSize(100);
            }
            result = preparedStatement.executeQuery();
            return com.gitee.melin.bee.util.JdbcUtils.handleResultSet(result, maxRecords);
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        } finally {
            JdbcUtils.closeResultSet(result);
            JdbcUtils.closeStatement(preparedStatement);
            JdbcUtils.closeConnection(conn);
        }
    }

    @Override
    public DataSourceType getDataSourceType() {
        return null;
    }

    @Override
    public long queryCount(String sql, Object... params) {
        String jdbcUrl = this.getJdbcUrl();
        Connection conn = null;
        PreparedStatement preparedStatement = null;
        ResultSet result = null;

        try {
            conn = DriverManager.getConnection(jdbcUrl, connectionInfo.getUsername(), connectionInfo.getPassword());

            preparedStatement = conn.prepareStatement(sql);
            if (MYSQL == this.getDataSourceType()) {
                preparedStatement.setFetchSize(Integer.MIN_VALUE);
            } else {
                preparedStatement.setFetchSize(100);
            }

            for (int i = 1; i <= params.length; i++) {
                preparedStatement.setObject(i, params[i - 1]);
            }

            result = preparedStatement.executeQuery();
            result.next();
            return result.getLong(1);
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        } finally {
            JdbcUtils.closeResultSet(result);
            JdbcUtils.closeStatement(preparedStatement);
            JdbcUtils.closeConnection(conn);
        }
    }

    @Override
    public List<Map<String, Object>> queryForList(String sql, Object... params) {
        String jdbcUrl = this.getJdbcUrl();
        if (SQLSERVER == this.getDataSourceType()) {
            jdbcUrl += ";selectMethod=cursor";
        }

        Connection conn = null;
        PreparedStatement preparedStatement = null;
        ResultSet result = null;
        try {
            conn = DriverManager.getConnection(jdbcUrl, connectionInfo.getUsername(), connectionInfo.getPassword());

            try {
                conn.setAutoCommit(false);
            } catch (SQLFeatureNotSupportedException ignore) {
            }

            preparedStatement = conn.prepareStatement(sql);
            if (MYSQL == this.getDataSourceType()) {
                preparedStatement.setFetchSize(Integer.MIN_VALUE);
            } else {
                preparedStatement.setFetchSize(100);
            }

            for (int i = 1; i <= params.length; i++) {
                preparedStatement.setObject(i, params[i - 1]);
            }

            result = preparedStatement.executeQuery();
            LinkedList<String> columnNames = Lists.newLinkedList();
            ResultSetMetaData metaData = result.getMetaData();
            int columnCount = metaData.getColumnCount();

            for (int i = 1; i <= columnCount; i++) {
                String columnName = metaData.getColumnLabel(i);
                columnNames.add(columnName);
            }

            List<Map<String, Object>> rows = Lists.newArrayList();
            while (result.next()) {
                Map<String, Object> row = Maps.newHashMap();
                for (String colName : columnNames) {
                    Object value = result.getObject(colName);
                    row.put(colName, value);
                }
                rows.add(row);
            }
            return rows;
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        } finally {
            JdbcUtils.closeResultSet(result);
            JdbcUtils.closeStatement(preparedStatement);
            JdbcUtils.closeConnection(conn);
        }
    }

    @Override
    public Boolean execute(String sql) {
        return execute(null, sql);
    }

    @Override
    public Boolean execute(String schema, String sql) {
        String jdbcUrl = this.getJdbcUrl();

        Connection conn = null;
        PreparedStatement preparedStatement = null;
        try {
            conn = DriverManager.getConnection(jdbcUrl, connectionInfo.getUsername(), connectionInfo.getPassword());

            if (schema != null) {
                if (MYSQL == connectionInfo.getDataSourceType() || STARROCKS == connectionInfo.getDataSourceType()) {
                    conn.setCatalog(schema);
                } else {
                    conn.setSchema(schema);
                }
            }

            conn.setAutoCommit(true);
            preparedStatement = conn.prepareStatement(sql);
            return preparedStatement.execute();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        } finally {
            JdbcUtils.closeStatement(preparedStatement);
            JdbcUtils.closeConnection(conn);
        }
    }

    @Override
    public Integer executeUpdate(String sql, Object... params) {
        return this.executeUpdate(null, sql, params);
    }

    @Override
    public Integer executeUpdate(SchemaIdentifier schema, String sql, Object... params) {
        String jdbcUrl = this.getJdbcUrl();

        Connection conn = null;
        PreparedStatement preparedStatement = null;
        try {
            conn = DriverManager.getConnection(jdbcUrl, connectionInfo.getUsername(), connectionInfo.getPassword());

            if (schema != null) {
                if (MYSQL == connectionInfo.getDataSourceType() || STARROCKS == connectionInfo.getDataSourceType()) {
                    conn.setCatalog(schema.getSchemaName());
                } else {
                    conn.setSchema(schema.getSchemaName());
                }
            }
            conn.setAutoCommit(true);
            preparedStatement = conn.prepareStatement(sql);
            for (int i = 1; i <= params.length; i++) {
                preparedStatement.setObject(i, params[i - 1]);
            }
            return preparedStatement.executeUpdate();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        } finally {
            JdbcUtils.closeStatement(preparedStatement);
            JdbcUtils.closeConnection(conn);
        }
    }

    protected BeeTableType getTableType(String type) {
        if (StringUtils.equalsIgnoreCase(type, "VIEW")) {
            return BeeTableType.VIEW;
        } else if (StringUtils.equalsIgnoreCase(type, "TABLE") || StringUtils.equalsIgnoreCase(type, "BASE TABLE")) {
            return BeeTableType.TABLE;
        } else if (StringUtils.equalsIgnoreCase(type, "Materialized View")) {
            return BeeTableType.MATERIALIZED_VIEW;
        } else if (StringUtils.equalsIgnoreCase(type, "EXTERNAL TABLE")) {
            return BeeTableType.EXTERNAL_TABLE;
        } else {
            throw new IllegalArgumentException("table type is invalid: " + type);
        }
    }

    protected boolean isParttitionTable(String partition) {
        return StringUtils.isNotBlank(partition);
    }

    private int interpretVersion(int result) {
        return result < 0 ? -9999 : result;
    }

    @Override
    public Statement getQueryStatement() {
        return preparedStatement;
    }

    protected abstract String getDDLColumnName(BeeTableType type);

    protected Driver loadDriver(String className) {
        try {
            Class<?> clazz = Class.forName(className);
            return (Driver) clazz.getDeclaredConstructor().newInstance();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage(), e);
        }
    }

    @Override
    public void close() {
        if (connectionFactory != null) {
            try {
                connectionFactory.close();
            } catch (Exception e) {
            }
        }
    }
}
